package de.pyranja.storage;

import java.io.InputStream;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;

import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;

import de.pyranja.storage.core.Staging;
import de.pyranja.storage.core.Storage;
import de.pyranja.storage.exception.BlobNotFoundException;
import de.pyranja.storage.support.ActionFactory;

public class AsyncBinaryStorage implements BinaryStorage {

	private final ListeningExecutorService dispatcher;
	private final ActionFactory create;
	private final Staging eden;
	private final Storage storage;
	
	public AsyncBinaryStorage(ExecutorService dispatcher, Staging eden,
			Storage storage, ActionFactory factory) {
		super();
		this.dispatcher = MoreExecutors.listeningDecorator(dispatcher);
		this.eden = eden;
		this.storage = storage;
		this.create = factory;
	}
	
	public void shutdown() throws InterruptedException {
		dispatcher.shutdown();
		dispatcher.awaitTermination(10, TimeUnit.SECONDS);
	}

	@Override
	public Blob store(InputStream data) {
		final Blob blob = eden.putUp(data);
		ListenableFuture<String> response = dispatcher.submit(create.save(blob.getId()));
		blob.setAsync(response);
		return blob;
	}

	@Override
	public InputStream fetch(String id) throws BlobNotFoundException {
		return storage.open(id);
	}

	@Override
	public ListenableFuture<String> claim(String id) throws BlobNotFoundException {
		ListenableFuture<String> response = dispatcher.submit(create.claim(id));
		return response;
	}

	@Override
	public ListenableFuture<String> release(String id) {
		ListenableFuture<String> response = dispatcher.submit(create.release(id));
		return response;
	}
}